/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2016-2026 hmc,Inc.All Rights Reserved.
 */
package com.sys.midware.MQ.rocketMQ.logic.impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.sys.midware.MQ.rocketMQ.logic.IPullConsumer;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * <pre>
*   IConsumer consumer = new ConsumerImpl();   
*   consumer.register(messageListener, topics);    
*   或者 
*   consumer.register(listener);
*   consumer.subscribe(topic, tags);
*   注意： 
*   consumer.start();  
*   consumer.shutdown();
 * </pre>
 * 
 * <b>ClassName：</b> ConsumerImpl <br/>
 * <b>Description：</b> rocketmq消费者类，使用示例 <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月19日 下午3:54:00 <br/>
 * <b>@version: </b> <br/>
 */
public class PullConsumerImpl implements IPullConsumer {

    private static final Logger logger = LoggerFactory.getLogger(PullConsumerImpl.class);

    // Java缓存
    private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();

    private final DefaultMQPullConsumer mqConsumer;


    /**
     * 最多取100条
     */
    private static final int maxNum = 100;

    /**
     * 
     * @param consumerGroup
     */
    public PullConsumerImpl(String consumerGroup) {
        this(consumerGroup, null);
    }

    public PullConsumerImpl(String nameServer, String consumerGroup) {
        mqConsumer = new DefaultMQPullConsumer(consumerGroup);
        mqConsumer.setNamesrvAddr(nameServer);
        mqConsumer.setInstanceName(consumerGroup);
    }

    public DefaultMQPullConsumer getMQConsumer() {
        return this.mqConsumer;
    }

    public void setNameServer(String namesrvAddr) {
        mqConsumer.setNamesrvAddr(namesrvAddr);
    }

    private static void putMessageQueueOffset(MessageQueue mq, long offset) {
        offseTable.put(mq, offset);
    }

    private static long getMessageQueueOffset(MessageQueue mq) {
        Long offset = offseTable.get(mq);
        if (offset != null) {
            return offset;
        }
        return 0;
    }

    @Override
    public void start() {
        try {
            logger.info(" ============= PullConsumer 准备启动 ==========");
            mqConsumer.start();
            logger.info(
                    " nameServer " + mqConsumer.getNamesrvAddr() + " , group " + mqConsumer.getConsumerGroup() + " ");
            logger.info(" ============= PullConsumer 启动成功 ==========");
        } catch (Exception e) {
            logger.error(" ============= PullConsumer start error!==========");
        }
    }

    @Override
    public void shutdown() {
        mqConsumer.shutdown();
    }

    @Override
    public MessageExt viewMessage(String msgId) {
        MessageExt ext = null;
        try {
            ext = mqConsumer.viewMessage(msgId);
        } catch (Exception e) {
            logger.error("viewMessage error", e);
        }
        return ext;
    }

    @Override
    public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) {
        Set<MessageQueue> mqs = null;
        try {
            mqs = mqConsumer.fetchSubscribeMessageQueues(topic);
        } catch (MQClientException e) {
            logger.error("fetchSubscribeMessageQueues error", e);
        }

        return mqs;
    }

    @Override
    public Set<String> getRegisterTopics() {
        Set<String> topics = null;
        try {
            topics = mqConsumer.getRegisterTopics();
            ;
        } catch (Exception e) {
            logger.error("getRegisterTopics error", e);
        }

        return topics;
    }

    @Override
    public List<MessageExt> queryMessageByTopicAndKey(String topic, String key) {
        List<MessageExt> extList = new ArrayList<MessageExt>();
        try {
            Set<MessageQueue> mqs = mqConsumer.fetchSubscribeMessageQueues(topic);
            for (MessageQueue mq : mqs) {
                logger.info("Consume from the queue: " + mq);
                QueryResult queryResult;
                try {
                    queryResult = mqConsumer.queryMessage(topic, key, maxNum, mqConsumer.minOffset(mq),
                            mqConsumer.maxOffset(mq));
                    extList.addAll(queryResult.getMessageList());
                } catch (InterruptedException e) {
                    logger.error("queryMessage error!MQ:" + mq, e);
                }
            }
        } catch (MQClientException e) {
            logger.error("pullMessageQueue error!", e);
        }

        return extList;
    }

    @Override
    public List<MessageExt> getMessageExtList(String topic) {
        List<MessageExt> msgList = new ArrayList<MessageExt>();
        try {
            Set<MessageQueue> mqs = mqConsumer.fetchSubscribeMessageQueues(topic);
            for (MessageQueue mq : mqs) {
                logger.info("Consume from the queue: " + mq);
                SINGLE_MQ: while (true) {
                    try {
                        PullResult pullResult = mqConsumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq),
                                maxNum);
                        logger.info("next BeginOffset:" + pullResult.getNextBeginOffset());
                        putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                        switch (pullResult.getPullStatus()) {
                        case FOUND:
                            List<MessageExt> list = pullResult.getMsgFoundList();
                            if (list != null && list.size() < 100) {
                                for (MessageExt msg : list) {
                                    msgList.add(msg);
                                }
                            }
                            break;
                        case NO_MATCHED_MSG:
                            break;
                        case NO_NEW_MSG:
                            break SINGLE_MQ;
                        case OFFSET_ILLEGAL:
                            break;
                        default:
                            break;
                        }
                    } catch (Exception e) {
                        logger.error("pullMessageQueue error! mq:" + mq, e);
                    }
                }
            }
        } catch (MQClientException e) {
            logger.error("pullMessageQueue error!", e);
        }
        return msgList;
    }

}
